package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        //流式处理消息
        // value是一个json格式的数据 key:xxxx value:{articleId:1001,UpdateArticleType:like,add:1}
        stream.map((key, value) ->{
            System.out.println(key +" : " + value);
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            //[key:1001,value:like:1]
            return new KeyValue<>(mess.getArticleId().toString(),mess.getType() +":" +mess.getAdd());
        })
        .groupBy((key, value) -> key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .aggregate(
                        /**
                         * 初始化：对结果进行初始化-这个方法在每个时间周期开始的时候执行
                         * 每个时间周期内执行一次
                         */
                        new Initializer<String>() {
                            @Override
                            public String apply() {
                                return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                            }
                        }
                        , new Aggregator<String, String, String>() {
                            /**
                             *
                             * @param key 当前消息的key -key:1001
                             * @param value 当前消息的value - value: like:1
                             * @param aggregate 上一次统计的结果
                             * @return
                             * 当前这个方式是每个消息来的时候都会执行一次
                             */
                            @Override
                            public String apply(String key, String value, String aggregate) {
                                if(StringUtils.isBlank(value)){
                                    return aggregate;
                                }

                                int col = 0,com=0,lik=0,vie=0;
                                //还原上一次统计的结果
                                String[] aggArg = aggregate.split(",");
                                for (String str : aggArg){ // str COLLECTION:0
                                    String[] split = str.split(":");
                                    switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                        case COLLECTION:
                                            col = Integer.parseInt(split[1]);
                                            break;
                                        case COMMENT:
                                            com = Integer.parseInt(split[1]);
                                            break;
                                        case LIKES:
                                            lik = Integer.parseInt(split[1]);
                                            break;
                                        case VIEWS:
                                            vie = Integer.parseInt(split[1]);
                                            break;
                                    }
                                }

                                //累加当前的结果
                                String[] split = value.split(":");
                                switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                    case COLLECTION:
                                        col += Integer.parseInt(split[1]);
                                        break;
                                    case COMMENT:
                                        com += Integer.parseInt(split[1]);
                                        break;
                                    case LIKES:
                                        lik += Integer.parseInt(split[1]);
                                        break;
                                    case VIEWS:
                                        vie += Integer.parseInt(split[1]);
                                        break;
                                }

                                String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                                System.out.println("文章的id:"+key);
                                System.out.println("当前时间窗口内的消息处理结果："+formatStr);
                                return formatStr;
                            }
                        }, Materialized.as("hot-atricle-stream-count-001")
                )
                .toStream()
                .map((key, value) -> {
                    /**
                     * key: 1001
                     * value: COLLECTION:1,COMMENT:10,LIKES:0,VIEWS:11
                     *
                     * * key: 1002
                     * value: COLLECTION:1,COMMENT:10,LIKES:0,VIEWS:11
                     */
                    return new KeyValue<>(key.key().toString(),formatObj(key.key().toString(),value.toString()));
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return stream;
    }

    private String formatObj(String key, String value) {
        ArticleVisitStreamMess streamMess = new ArticleVisitStreamMess();
        streamMess.setArticleId(Long.parseLong(key));

        String[] split = value.split(",");

        for (String str : split) {
            String[] strs = str.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(strs[0])){
                case COLLECTION:
                    streamMess.setCollect(Integer.parseInt(strs[1]));
                    break;
                case COMMENT:
                    streamMess.setComment(Integer.parseInt(strs[1]));
                    break;
                case LIKES:
                    streamMess.setLike(Integer.parseInt(strs[1]));
                    break;
                case VIEWS:
                    streamMess.setView(Integer.parseInt(strs[1]));
                    break;
            }
        }

        return JSON.toJSONString(streamMess);
    }
}
